package org.eclipse.jetty.deploy;

import java.io.BufferedInputStream;
import java.io.BufferedOutputStream;
import java.io.DataInputStream;
import java.io.File;
import java.io.FileOutputStream;
import java.io.OutputStream;
import java.net.URI;
import java.util.Iterator;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FileSystem;
import org.eclipse.jetty.deploy.AppProvider;
import org.eclipse.jetty.deploy.DeploymentManager;
import org.eclipse.jetty.deploy.graph.Node;
import org.eclipse.jetty.deploy.graph.Path;
import org.eclipse.jetty.deploy.providers.WebAppProvider;
import org.eclipse.jetty.util.log.Log;

/**
 * @author haruyosi
 */
public class HdfsDeploymentManager extends DeploymentManager {

	/**
	 * @param appId
	 * @param nodeName
	 * @param userName
	 */
	public void requestAppGoal(String appId, String nodeName) {

		AppEntry appentry = findAppByOriginId(appId);
		if (appentry == null) {
			if (appId.startsWith("hdfs:")) {
				WebAppProvider webap = null;
				for (AppProvider ap : super.getAppProviders()) {
					if (ap instanceof WebAppProvider) {
						webap = (WebAppProvider) ap;
						break;
					}
				}
				if (webap == null) {
					throw new IllegalStateException(
							"App not being tracked by Deployment Manager: "
									+ appId);
				}
				if ("starting".equals(nodeName)) {
					copyFromHdfs(webap, appId);
				} else if ("undeployed".equals(nodeName)) {
					deleteFromLocal(webap, appId);
				}
				return;
			} else {
				throw new IllegalStateException(
						"App not being tracked by Deployment Manager: " + appId);
			}
		}
		requestAppGoal(appentry, nodeName);
	}

	private void deleteFromLocal(WebAppProvider webap, String originId) {
		try {
			URI uri = new URI(webap.getMonitoredDirName());
			String sDfsFile = originId.substring(originId.lastIndexOf("/"));
			File deleteFile = new File(uri.getPath() + sDfsFile);
			if (deleteFile.exists() && deleteFile.isFile()) {
				deleteFile.delete();
			}
		} catch (Exception e) {
			e.printStackTrace();
			throw new IllegalStateException(e);
		}
	}

	/**
	 * @param webap
	 * @param originId
	 */
	private void copyFromHdfs(WebAppProvider webap, String originId) {
		try {
			URI uri = new URI(webap.getMonitoredDirName());

			Configuration conf = new Configuration();
			// String sDfsPath = originId.substring(0,
			// originId.lastIndexOf("/"));
			String sDfsFile = originId.substring(originId.lastIndexOf("/"));
			URI orgUri = new URI(originId);
			String sUserName = orgUri.getUserInfo().substring(0,
					orgUri.getUserInfo().indexOf(":"));
			URI dfsUri = new URI(originId.replace(orgUri.getUserInfo() + "@",
					""));
			FileSystem fs = FileSystem.get(dfsUri, conf, sUserName);
			org.apache.hadoop.fs.Path dfsPath = new org.apache.hadoop.fs.Path(
					dfsUri);
			if (fs.isFile(dfsPath)) {
				OutputStream bout = new BufferedOutputStream(
						new FileOutputStream(uri.getPath() + sDfsFile));
				DataInputStream dis = fs.open(dfsPath);
				BufferedInputStream bis = new BufferedInputStream(dis);
				byte[] buf = new byte[4096];
				int length;
				while (-1 < (length = bis.read(buf))) {
					bout.write(buf, 0, length);
				}
				bout.flush();
				bout.close();
				bis.close();
			}
		} catch (Exception e) {
			e.printStackTrace();
			throw new IllegalStateException(e);
		}
	}

	/**
	 * @param originId
	 * @return
	 */
	private AppEntry findAppByOriginId(String originId) {

		if (originId == null) {
			return null;
		}

		for (AppEntry entry : super.getAppEntries()) {
			if (originId.equals(entry.getApp().getOriginId())) {
				return entry;
			}
		}
		return null;
	}

	/**
	 * @param appentry
	 * @param nodeName
	 */
	private void requestAppGoal(AppEntry appentry, String nodeName) {

		Node destinationNode = super.getLifeCycle().getNodeByName(nodeName);
		if (destinationNode == null) {
			throw new IllegalStateException(
					"Node not present in Deployment Manager: " + nodeName);
		}
		// Compute lifecycle steps
		Path path = super.getLifeCycle().getPath(appentry.getLifecyleNode(),
				destinationNode);
		if (path.isEmpty()) {
			// nothing to do. already there.
			return;
		}

		// Execute each Node binding. Stopping at any thrown exception.
		try {
			Iterator<Node> it = path.getNodes().iterator();
			if (it.hasNext()) // Any entries?
			{
				// The first entry in the path is always the start node
				// We don't want to run bindings on that entry (again)
				it.next(); // skip first entry
				while (it.hasNext()) {
					Node node = it.next();
					Log.getLog().debug("Executing Node: " + node);
					super.getLifeCycle().runBindings(node, appentry.getApp(),
							this);
					appentry.setLifeCycleNode(node);
				}
			}
		} catch (Throwable t) {
			Log.getLog().warn("Unable to reach node goal: " + nodeName, t);
		}
	}

}
